package com.wuwangfu.topn;

import com.wuwangfu.entity.EventBean;
import com.wuwangfu.func.UrlsAggregateFunc01;
import com.wuwangfu.func.UrlsProcessWindowFunc01;
import com.wuwangfu.source.CustomEventSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-07-06
 *
 * 要统计10秒内访问量最多的5条url，5秒钟刷新一次
 * 分析题目，10秒内的数据进行统计，5秒钟刷新一次，首先确定是滑动窗口
 *
 * 第一个方案（差）：
 * 流水线使用乱序+延迟，计算则不用分区，针对10秒内所有的数据，使用增量函数+全窗口函数，
 * 统计所有url的访问次数，对其所有的url的访问次数进行排序，再输出前5条
 */
public class HotUrlsTopN01 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setAutoWatermarkInterval(200);
        env.addSource(new CustomEventSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.
                        <EventBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((SerializableTimestampAssigner<EventBean>)
                                (event,l)->event.getTimestamp())
                ).keyBy(event->true)
                .window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))
                .aggregate(new UrlsAggregateFunc01(),new UrlsProcessWindowFunc01())
                .print();

        env.execute();

    }
}
